package com.godyao.mall.rabbitmq.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.godyao.mall.common.utils.AssertUtils;
import com.godyao.mall.rabbitmq.model.MqMsg;
import com.godyao.mall.rabbitmq.repository.MqMsgRepository;
import com.godyao.mall.rabbitmq.service.ISendMsgService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * @author godyao
 * @date 2022/4/5
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class SendMsgServiceImpl implements ISendMsgService {
    private final RabbitTemplate rabbitTemplate;

    private final MqMsgRepository mqMsgRepository;

    @PostConstruct
    public void init() {
        //消息未送达队列触发回调
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            final MqMsg msg;
            try {
                log.error("消息发送失败，未送达队列，message：{},replyCode：{}，replyText：{}，exchange：{}，exchange：{}", new ObjectMapper().writeValueAsString(message), replyCode, replyText, exchange, routingKey);
                msg = new ObjectMapper().readValue(new String(message.getBody()), MqMsg.class);
            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }
            //设置消息的状态为发送失败
            MqMsg msg1 = MqMsg.builder()
                    .id(msg.getId())
                    //设置状态为未送达队列
                    .status(3)
                    .build();
            mqMsgRepository.save(msg1);
        });
        //消息进入到Exchange触发回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            String id = Objects.requireNonNull(correlationData.getId());
            if (!ack) {
                log.error("消息未发送成功，返回信息：{}", cause);
                //设置消息的状态为发送失败
                MqMsg msg1 = MqMsg.builder()
                        .id(Long.valueOf(id))
                        .status(0)
                        .build();
                mqMsgRepository.save(msg1);
            } else { //消息发送成功
                //设置消息的状态为发送成功
                MqMsg msg1 = MqMsg.builder()
                        .id(Long.valueOf(id))
                        .status(2)
                        .build();
                mqMsgRepository.save(msg1);
            }
        });
    }

    @Override
    public void sendMsg(String exchange, String routingKey, String data, Integer delayTime) {
        // 消息入库 返回唯一ID
        final MqMsg msg = MqMsg.builder()
                .createTime(new Date())
                .exchange(exchange)
                .routingKey(routingKey)
                .msg(data)
                .updateTime(new Date())
                .status(1)
                .build();
        final MqMsg save = mqMsgRepository.save(msg);
        AssertUtils.assertTrue(ObjectUtil.isNotNull(save));
        // 组装消息内容
        buildAndSendMsg(msg, data, exchange, routingKey, delayTime);
    }

    @Override
    public void reSendMsg() {
        // 查询消息表中未发送成功的消息 查找失败5次以及5次之内的消息
        final List<MqMsg> mqMsgs = mqMsgRepository.listNoSendSuccess();
        if (CollectionUtils.isEmpty(mqMsgs)) return;
        for (MqMsg msg : mqMsgs) {
            buildAndSendMsg(msg, msg.getMsg(), msg.getMsg(), msg.getRoutingKey(), null);
        }
    }

    /**
     * 组装并且发送消息
     * @param msg
     * @param msg2
     * @param msg3
     * @param routingKey
     */
    private void buildAndSendMsg(MqMsg msg, String msg2, String msg3, String routingKey, Integer delayTime) {
        // 组装消息内容
        final MessageProperties messageProperties = new MessageProperties();
        // 消息唯一ID，防止幂等问题
        messageProperties.setMessageId(msg.getId().toString());
        messageProperties.setDelay(delayTime);
        final Message message = new Message(msg2.getBytes(StandardCharsets.UTF_8), messageProperties);
        // 发送消息，利用CorrelationDaa将ID传递下去
        //这里不必try-catch 即使发送失败
        rabbitTemplate.convertAndSend(msg3, routingKey, message, new CorrelationData(msg.getId().toString()));
    }
}
